/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/
package org.apache.james.mailbox.cassandra.mail;


import java.util.UUID;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.cassandra.utils.ByteBufferUtil;

import org.apache.james.mailbox.MailboxException;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.cassandra.CNames;
import org.apache.james.mailbox.cassandra.CUtils;
import org.apache.james.mailbox.store.mail.ModSeqProvider;
import org.apache.james.mailbox.store.mail.model.Mailbox;
import org.apache.openjpa.lib.util.Bytes;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

/**
 * ModSeqProvider implementation for HBase.
 * 
 */
public class CModSeqProvider implements ModSeqProvider<UUID> {

    /** Link to the HBase Configuration object and specific mailbox names */
    
   
@Override
    
public long highestModSeq(MailboxSession session, Mailbox<UUID> mailbox) throws MailboxException {
   
    long modSeq = 0;

      
     try{
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);  
        
        byte[] key=CUtils.mailboxRowKey(mailbox.getMailboxId());
        String k=key.toString();
        
        ColumnPath path = new ColumnPath(ByteBufferUtil.string(CNames.MAILBOX_CF));
        path.setColumn(CNames.MAILBOX_HIGHEST_MODSEQ);
	
        ColumnOrSuperColumn result = client.get(ByteBufferUtil.bytes(k),path,ConsistencyLevel.ONE);
 	Column cname =  result.getColumn();
             
        byte[] cval=cname.getValue();
        modSeq = Bytes.toLong(cval);
                 
       
                
               
        }catch(Exception e){
            
        }
        
         return modSeq;
    
    
}
    @Override
    public long nextModSeq(MailboxSession session, Mailbox<UUID> mailbox) throws MailboxException {
        
    long modSeq = 0;

      
     try{
        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
        TProtocol protocol = new TBinaryProtocol(transport);
        transport.open();                 
        Cassandra.Client client = new Cassandra.Client(protocol);
        client.set_keyspace(CNames.keyspace);  
        
        byte[] key=CUtils.mailboxRowKey(mailbox.getMailboxId());
        String k=key.toString();
        
        ColumnPath path = new ColumnPath(ByteBufferUtil.string(CNames.MAILBOX_CF));
        path.setColumn(CNames.MAILBOX_HIGHEST_MODSEQ);
	
        ColumnOrSuperColumn result = client.get(ByteBufferUtil.bytes(k),path,ConsistencyLevel.ONE);
 	Column cname =  result.getColumn();
             
        byte[] cval=cname.getValue();
        modSeq = Bytes.toLong(cval);
       
        }catch(Exception e){
            
        }
        
         return modSeq;
           
        }
 }
